﻿using RockBus.Logging.Interfaces;
using RockBus.MessageRepository.Interfaces;
using RockBus.Messages.Interfaces;
using RockBus.PerformanceCounters.Interfaces;
using RockBus.Routing.Interfaces;
using RockBus.Utilities;
using System;
using System.Reflection;
using System.ServiceModel.Channels;
using System.Threading;
using System.Xml;

namespace RockBus.MessageProcessors.MessageRepository
{
    public class MessagePublisher : IMessagePublisher
    {
        public const string QueueEntryIdKey = "QueueEntryId";

        private const string OperationCounterName = "PublishMessage";

        public MessagePublisher(
            ILogger logger,
            IPerformanceCounterFactory performanceCounterFactory,
            IMessageFactory messageFactory,
            IQueueManager queueManager,
            IMessageRepositoryHelper messageRepositoryHelper)
        {
            this.Logger = logger;
            this.PerformanceCounterFactory = performanceCounterFactory;
            this.MessageFactory = messageFactory;
            this.QueueManager = queueManager;
            this.MessageRepositoryHelper = messageRepositoryHelper;

            this.XmlReaderSettings.CloseInput = false;
            this._pollTimer = new Lazy<Timer>(() =>
            {
                return new Timer(new TimerCallback(this.PollTimerCallback));
            }, true);
        }

        private ILogger Logger { get; set; }

        private IPerformanceCounterFactory PerformanceCounterFactory { get; set; }

        private IMessageFactory MessageFactory { get; set; }

        private IQueueManager QueueManager { get; set; }

        private IMessageRepositoryHelper MessageRepositoryHelper { get; set; }

        private readonly XmlReaderSettings XmlReaderSettings = new XmlReaderSettings();

        private readonly Lazy<Timer> _pollTimer;

        private Timer PollTimer { get { return this._pollTimer.Value; } }

        private bool _enabled;
        private readonly object _enabledLock = new object();

        public void Enable()
        {
            lock (this._enabledLock)
            {
                if (this._enabled)
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "DbMessagePublisher was already enabled.");
                }
                else
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Enabling DbMessagePublisher.");
                    this.PollTimer.Change(0, this.MessageRepositoryHelper.PollingIntervalMs);
                    this._enabled = true;
                }
            }
        }

        public void Disable()
        {
            lock (this._enabledLock)
            {
                if (!this._enabled)
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "DbMessagePublisher was already disabled.");
                }
                else
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Disabling DbMessagePublisher.");
                    this.PollTimer.Change(Timeout.Infinite, Timeout.Infinite);
                    this._enabled = false;
                }
            }
        }

        private int _runningGuard;

        private void PollTimerCallback(object state)
        {
            try
            {
                if (Interlocked.CompareExchange(ref _runningGuard, 1, 0) == 1)
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "PollTimerCallback was ignored because another was already running.");
                    return;
                }

                this.DequeueAndProcessAsync();
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);

                // DO NOT throw HERE! We need to catch ANY exception on the Timer.
                //throw;
            }
        }

        private async void DequeueAndProcessAsync()
        {
            try
            {
                var dequeueResult = await this.MessageRepositoryHelper.StartDequeueEntries();

                this.ProcessDequeueResult(dequeueResult);
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
            }
            finally
            {
                _runningGuard = 0;
            }
        }

        private void ProcessDequeueResult(DequeueResult dequeueResult)
        {
            if ((null == dequeueResult) || (null == dequeueResult.QueueEntries) || (dequeueResult.QueueEntries.Count <= 0))
            {
                //this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Dequeued zero QueueEntries.");
                return;
            }

            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Dequeued {0} QueueEntries.", dequeueResult.QueueEntries.Count);

            foreach (var qe in dequeueResult.QueueEntries)
            {
                this.ProcessQueueEntry(qe);
            }
        }

        private void ProcessQueueEntry(QueueEntry qe)
        {
            var oc = this.PerformanceCounterFactory.CreateOperationCounter(OperationCounterName);
            oc.Start();

            try
            {
                var messageVersion = Conversions.ConvertWithTypeConverter<string, MessageVersion>(qe.MessageVersion);

                var messageStream = new ReadableDechunkingStream(this.Logger, qe.Size, qe.PartCount, chunkIndex =>
                {
                    return this.MessageRepositoryHelper.GetMessageFragment(qe.MessageId, chunkIndex);
                });
                var reader = XmlReader.Create(messageStream, this.XmlReaderSettings);
                var message = Message.CreateMessage(reader, 4096, messageVersion);

                var messageContext = this.MessageFactory.CreateMessageContext(message, qe.RoutingHeader);
                messageContext.SetProperty(MessagePublisher.QueueEntryIdKey, qe.Id);

                // Increase the MessageDirection (Inbound -> outbound)
                messageContext.ProcessingData.MessageDirection++;

                this.QueueManager.ProcessMessage(messageContext);
            }
            catch (Exception exc)
            {
                oc.Abort();
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
            }
            finally
            {
                oc.Complete();
            }
        }
    }
}